package com.ysy.mq.collect;

import com.alibaba.fastjson2.JSON;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.ysy.constant.RocketMqConstant;
import com.ysy.entity.NoteCollect;
import com.ysy.entity.dto.NoteCollectDTO;
import com.ysy.mapper.NoteCollectMapper;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Objects;

;

@Component
@RocketMQMessageListener(consumerGroup = "persimmons_group_"+ RocketMqConstant.NOTE_COLLECT_TOPIC,
        topic = RocketMqConstant.NOTE_COLLECT_TOPIC,
        consumeMode = ConsumeMode.ORDERLY)
public class NoteCollectConsumer implements RocketMQListener<Message> {

    RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private NoteCollectMapper noteCollectMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();
        String body = new String(message.getBody());
        NoteTagMessageHandle(body);
    }

    private void NoteTagMessageHandle(String body) {
        NoteCollectDTO noteCollectDBDTO = JSON.parseObject(body, NoteCollectDTO.class);
        if (Objects.isNull(noteCollectDBDTO)) return;
        // 执行落库
        NoteCollect collect = new NoteCollect();
        collect.setUserId(noteCollectDBDTO.getUserId());
        collect.setNoteId(noteCollectDBDTO.getNoteId());
        collect.setCreateTime(noteCollectDBDTO.getCreateTime());
        collect.setStatus(noteCollectDBDTO.getType());
      //  boolean result = noteLikeMapper.insertOrUpdate(noteLike); 效率低查询2次
        int count = noteCollectMapper.insertOrUpdateSql(collect);
        if (count == 0) return;
        // 落库成功进行mq计数
        MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(body);
        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(RocketMqConstant.NOTE_COLLECT_COUNT_TOPIC, messageBuilder, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
            }

            @Override
            public void onException(Throwable throwable) {
            }
        });
    }
}
